import base64
import json
import urllib.parse
from abc import abstractmethod
from collections import Counter, defaultdict
from io import BytesIO
from pathlib import Path
from typing import Any

from multipart import parse_form as orig_parse_form
from multipart.multipart import Field, File

from ..models.base import Extensible
from ..models.suite import File as FileModel
from ..models.suite import ReqInfo, TestCase, TestStep, TestSuite
from ..utils import str_utils
from ..utils.exc_utils import exc_scope, exc_summary, exc_tracker


class StepGenerator:
    """避免重复生成相同名称的步骤"""

    def __init__(self):
        self.step_name_counter = defaultdict(int)

    def gen_step_name(self, path: str) -> str:
        """传入接口路径，生成测试步骤的名称，如果有重复接口，名称后会带上"_重复次数"

        :param path: 接口路径
        :return: 生成的步骤名称，默认将接口路径转为蛇形命名
        """
        final_name = step_name = path[1:].replace("/", "_").replace("-", "_")
        if num := self.step_name_counter[step_name]:
            final_name = f"{step_name}_{num}"

        self.step_name_counter[step_name] += 1
        return final_name


IGNORED_HEADERS = [
    "Host",
    "Content-Length",
    "User-Agent",
    "Origin",
    "Upgrade-Insecure-Requests",
    "Cookie",
    "Connection",
    "Cache-Control",
    "Pragma",
    "Content-Type",
    "Authorization",
    "Sec-Fetch-User",
    "Sec-Fetch-Site",
    "Sec-Fetch-Mode",
    "Sec-Fetch-Dest",
    "Accept-Encoding",
    "Accept-Language",
    "sec-ch-ua",
    "sec-ch-ua-mobile",
    "sec-ch-ua-platform",
    "Accept",
    "X-Requested-With",
]

def unpack_single_item_lists(data: dict[Any, list[Any]]):
    """
    如果dict中某个键的值是单元素list，则直接拆开
    eg: {"a": [1]} -> {"a": 1}
    """
    return {
        k: (v if isinstance(v, list) and len(v) > 1 else v[0]) for k, v in data.items()
    }
def header_list2dict(headers: list[dict[str, str]]):
    return {header["name"]: header["value"] for header in headers}


def list2dict(lst: list[dict[str, str]], key="name", value="value"):
    new_dict = defaultdict(list)
    for item in lst:
        new_dict[item[key]].append(item[value])
    return unpack_single_item_lists(dict(new_dict))


def count_and_replace_base_url(steps: list[TestStep]):
    """如果没有指定 base_url，则将使用最多的 base_url 设为默认值（None），同时替换 Referer"""
    most_common_base_url = Counter(
        step.req_info.base_url for step in steps
    ).most_common(1)[0][0]

    for step in steps:
        if step.req_info.base_url == most_common_base_url:
            step.req_info.base_url = None
            if (referer := step.req_info.headers.get("Referer")) and referer.startswith(
                str(most_common_base_url)
            ):
                step.req_info.headers["Referer"] = referer.replace(
                    str(most_common_base_url), "${base_url}"
                )


def parse_form(
    headers: dict[str, str],
    stream: BytesIO,
    chunk_size: int = 1048576,
) -> dict[str, list[Any]]:
    items: dict[str, list[Any]] = defaultdict(list)

    def on_field(field: Field):
        if not (isinstance(field.field_name, bytes) and isinstance(field.value, bytes)):
            raise TypeError("什么，读取到的值竟然不是 bytes？")  # impossible
        name = field.field_name.decode()
        value = field.value.decode()
        items[name].append(value)

    def on_file(file: File):
        if not isinstance(file.field_name, bytes):
            raise TypeError("什么，读取到的值竟然不是bytes？")  # impossible
        name = file.field_name.decode()
        # 需要将文件offset重置为0，否则无内容可读
        file.file_object.seek(0)
        value = FileModel(
            filename=file.file_name,
            base64=base64.b64encode(file.file_object.read()).decode(),
        ).model_dump()
        items[name].append(value)

    orig_parse_form(
        headers=headers,
        input_stream=stream,
        on_field=on_field,
        on_file=on_file,
        chunk_size=chunk_size,
    )

    return items


class SuiteConverter(Extensible, base=True):
    """用例转换器基类，将指定类型的抓包内容转为 TestSuite

    如要自定义，请继承并实现 convert 方法，并通过 type 指定文件后缀名"""

    @classmethod
    @abstractmethod
    def convert(
        cls,
        capture: Path | str,
        name: str | None = None,
        suite_type: str | None = None,
        base_url: str | None = None,
    ) -> TestSuite:
        """将抓包内容转为TestSuite

        :param capture: 抓包内容，可以是文件路径或字符串
        :param name: 抓包的名称
        :param base_url: 若为空，则将以 origin 作为每个 TestStep 的 base_url
        :return: 测试套件
        """
        pass


class ChlsjSuiteConverter(SuiteConverter, type="chlsj"):

    @staticmethod
    @exc_tracker
    def parse_request_body(entry: dict[str, Any]):
        """解析请求报文"""
        request = entry["request"]
        body: dict[str, Any] | None = request.get("body")
        mime_type: str | None = request["mimeType"]
        data = None
        if body and mime_type:
            charset = body.get("charset") or "iso-8859-1"
            text: str | None = body.get("text")
            if "json" in mime_type:
                assert isinstance(text, str)  # please stupid pylance 🙄
                text = text.encode(charset).decode("utf-8", "ignore")
                try:
                    data = json.loads(text)
                except json.JSONDecodeError as e:
                    raise ValueError(
                        f"请求体不是 json，转换失败：\n{exc_summary(e)}"
                    ) from e
            elif mime_type == "text/plain":
                data = text
            elif mime_type == "application/x-www-form-urlencoded":
                data = (
                    unpack_single_item_lists(
                        urllib.parse.parse_qs(text, keep_blank_values=True)
                    )
                    if text
                    else None
                )
            elif mime_type == "multipart/form-data":
                data = defaultdict(list)
                if isinstance(text, str):  # please stupid pylance 🙄
                    data_bytes = text.encode(charset)
                elif "encoded" in body:
                    match body["encoding"]:
                        case "base64":
                            data_bytes = base64.b64decode(body["encoded"])
                        case _ as encoding:
                            raise NotImplementedError(
                                f"上传文件暂未支持 encoding: {encoding}"
                            )
                else:
                    raise NotImplementedError(f"未知的 body 格式：{body}")
                data = unpack_single_item_lists(
                    parse_form(
                        headers=header_list2dict(entry["request"]["header"]["headers"]),
                        stream=BytesIO(data_bytes),
                    )
                )
            else:
                raise NotImplementedError(f"暂未支持 mimeType: {mime_type}")
        return data

    @classmethod
    def convert(
        cls,
        capture: Path | str,
        name: str | None = None,
        suite_type: str | None = None,
        base_url: str | None = None,
    ) -> TestSuite:
        content = capture.read_text() if isinstance(capture, Path) else capture
        step_generator = StepGenerator()
        steps: list[TestStep] = []
        for index, entry in enumerate(json.loads(content)):

            port = entry["actualPort"]
            if (
                entry["scheme"] == "https"
                and port == 443
                or entry["scheme"] == "http"
                and port == 80
            ):
                port_part = ""
            else:
                port_part = f":{port}"
            url = f"{entry['scheme']}://{entry['host']}{port_part}{entry['path']}"

            with exc_scope(f"将第 {index} 个请求（{url}）转为 TestStep"):

                # 传入了 base_url，且 url 是以 base_url 开头，则设置本步骤采用传入的 base_url
                if base_url and url.startswith(base_url):
                    step_base_url, path = None, url.removeprefix(base_url)
                # 否则，本步骤的 base_url 为 origin（即去掉 path）
                else:
                    step_base_url = f"{entry['scheme']}://{entry['host']}{port_part}"
                    path: str = entry["path"] or ""

                step = exc_tracker(
                    TestStep if not suite_type else TestStep.get(suite_type)
                )(
                    step_name=step_generator.gen_step_name(path),
                    step_desc="步骤描述_不改的人是🐷",
                    req_info=ReqInfo(
                        base_url=step_base_url,
                        method=entry["method"],
                        mime_type=entry["request"]["mimeType"],
                        path=path,
                        headers={
                            k: v
                            for k, v in header_list2dict(
                                entry["request"]["header"]["headers"]
                            ).items()
                            if k not in IGNORED_HEADERS
                        },
                    ),
                    test_cases=[
                        (TestCase if not suite_type else TestCase.get(suite_type))(
                            params=(
                                unpack_single_item_lists(
                                    urllib.parse.parse_qs(query, keep_blank_values=True)
                                )
                                if (query := entry["query"])
                                else None
                            ),
                            data=cls.parse_request_body(entry),
                        )
                    ],
                )
                if step:
                    steps.append(step)

        if not base_url:
            count_and_replace_base_url(steps)

        return (TestSuite if not suite_type else TestSuite.get(suite_type))(
            suite_name=str_utils.snake2pascal(
                name.split(".")[0] if name else "套件名称"
            ),
            suite_desc="套件描述_不改的人是🐷",
            test_steps=list(steps),  # please stupid pylance 🙃
        )


class HARSuiteConverter(SuiteConverter, type="har"):

    @staticmethod
    def parse_post_data(post_data: dict[str, Any]) -> Any:
        mime_type = post_data["mimeType"].split(";")[0].strip()

        match mime_type:
            case "text/plain":
                return post_data["text"]
            case "application/x-www-form-urlencoded":
                return unpack_single_item_lists(list2dict(post_data["params"]))
            case "application/json":
                try:
                    return json.loads(post_data["text"])
                except json.JSONDecodeError as e:
                    raise ValueError(
                        f"请求体不是 json，转换失败：\n{exc_summary(e)}"
                    ) from e
            case "multipart/form-data":
                return {param["name"]: param["value"] for param in post_data["params"]}
            case _:
                raise NotImplementedError(f"未知的请求体类型：{mime_type}")

    @classmethod
    def convert(
        cls,
        capture: Path | str,
        name: str | None = None,
        suite_type: str | None = None,
        base_url: str | None = None,
    ) -> TestSuite:
        content = capture.read_text() if isinstance(capture, Path) else capture
        step_generator = StepGenerator()
        steps: list[TestStep] = []
        for index, entry in enumerate(
            filter(
                # 仅保留 xhr / fetch / document 类型的请求
                lambda x: x["_resourceType"] in ["xhr", "fetch", "document"],
                json.loads(content)["log"]["entries"],
            )
        ):
            request = entry["request"]
            url: str = request["url"]
            parsed_url = urllib.parse.urlparse(url)
            query = parsed_url.query

            with exc_scope(f"将第 {index} 个请求 {url} 转为 TestStep"):
                if base_url and url.startswith(base_url):
                    step_base_url = None
                    path = url.removeprefix(base_url)
                    if query:
                        path = path[: path.find("?")]
                else:
                    path = urllib.parse.urlparse(url).path
                    step_base_url = (
                        url[: url.find("?")] if query else url
                    ).removesuffix(path)

                headers = header_list2dict(request["headers"])
                step = exc_tracker(
                    TestStep if not suite_type else TestStep.get(suite_type)
                )(
                    step_name=step_generator.gen_step_name(path),
                    step_desc="步骤描述_不改的人是🐷",
                    req_info=ReqInfo(
                        method=request["method"],
                        base_url=step_base_url,
                        path=path,
                        mime_type=(
                            entry["postData"]["mimeType"].split(";")[0].strip()
                            if "postData" in entry
                            else None
                        ),
                        headers={
                            k: v for k, v in headers.items() if k not in IGNORED_HEADERS
                        },
                    ),
                    test_cases=[
                        (TestCase if not suite_type else TestCase.get(suite_type))(
                            params=(
                                unpack_single_item_lists(
                                    urllib.parse.parse_qs(query, keep_blank_values=True)
                                )
                                if query
                                else None
                            ),
                            data=entry["postData"] if "postData" in entry else None,
                        )
                    ],
                )

                if step:
                    steps.append(step)

        if not base_url:
            count_and_replace_base_url(steps)

        return (TestSuite if not suite_type else TestSuite.get(suite_type))(
            suite_name=str_utils.snake2pascal(
                name.split(".")[0] if name else "套件名称"
            ),
            suite_desc="套件描述_不改的人是🐷",
            test_steps=list(steps),
        )
